package org.apache.rocketmq.proxy.grpc.v2.channel;

import apache.rocketmq.v2.*;
import com.google.common.base.MoreObjects;
import com.google.common.collect.ComparisonChain;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.TextFormat;
import com.google.protobuf.util.JsonFormat;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.common.channel.ChannelHelper;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
import org.apache.rocketmq.proxy.processor.channel.ChannelExtendAttributeGetter;
import org.apache.rocketmq.proxy.processor.channel.ChannelProtocolType;
import org.apache.rocketmq.proxy.processor.channel.RemoteChannel;
import org.apache.rocketmq.proxy.processor.channel.RemoteChannelConverter;
import org.apache.rocketmq.proxy.service.relay.ProxyChannel;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayResult;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
import org.apache.rocketmq.proxy.service.transaction.TransactionData;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.header.CheckTransactionStateRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ConsumeMessageDirectlyResultRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;

public class GrpcClientChannel extends ProxyChannel implements ChannelExtendAttributeGetter, RemoteChannelConverter {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
    //把GrpcChannelManager管理器定义在这里了
    private final GrpcChannelManager grpcChannelManager;

    //grpc客户端设置管理器
    private final GrpcClientSettingsManager grpcClientSettingsManager;
    //这里定义了一个原子引用对象，保存了StreamObserver流式观察者对象。我之前跟大家说proxy的服务端可以使用
    //为客户端创建的channel和客户端通信，在这里就得到印证了，GrpcClientChannel对象持有了可以和客户端进行通信的GrpcClientChannel
    //所以可以随时和客户端进行通信
    private final AtomicReference<StreamObserver<TelemetryCommand>> telemetryCommandRef = new AtomicReference<>();

    private final Object telemetryWriteLock = new Object();

    //当前客户端Channel对应的客户端ID
    private String clientId;

    public GrpcClientChannel(ProxyRelayService proxyRelayService, GrpcClientSettingsManager grpcClientSettingsManager,
                             GrpcChannelManager grpcChannelManager, ProxyContext ctx, String clientId) {
        super(proxyRelayService, null, new GrpcChannelId(clientId),
                ctx.getRemoteAddress(),
                ctx.getLocalAddress());

        this.grpcChannelManager = grpcChannelManager;
        this.grpcClientSettingsManager = grpcClientSettingsManager;
        this.clientId = clientId;
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到当前客户端channel扩展属性的方法，其实就是得到channel对应的客户端的设置信息
     */
    @Override
    public String getChannelExtendAttribute() {
        Settings settings = this.grpcClientSettingsManager.getRawClientSettings(this.clientId);
        if (settings == null) {
            return null;
        }
        try {
            return JsonFormat.printer().print(settings);
        } catch (InvalidProtocolBufferException e) {
            log.error("convert settings to json data failed. settings:{}", settings, e);
        }
        return null;
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：解析客户端channel扩展属性的方法，和上面方法不同的是，该方法会把channel对应的客户端的设置信息设置到Settings对象中，并返回这个对象
     */
    public static Settings parseChannelExtendAttribute(Channel channel) {
        if (ChannelHelper.getChannelProtocolType(channel).equals(ChannelProtocolType.GRPC_V2) &&
                channel instanceof ChannelExtendAttributeGetter) {
            String attr = ((ChannelExtendAttributeGetter) channel).getChannelExtendAttribute();
            if (attr == null) {
                return null;
            }

            Settings.Builder builder = Settings.newBuilder();
            try {
                JsonFormat.parser().merge(attr, builder);
                return builder.build();
            } catch (InvalidProtocolBufferException e) {
                log.error("convert settings json data to settings failed. data:{}", attr, e);
                return null;
            }
        }
        return null;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把当前客户端channel转换成RemoteChannel的方法，现在大家还不清楚RemoteChannel的作用，还有很久我们才会用到，但现在我可以为大家提前简单解释一下
     * 假如我们构建的是Proxy集群，那么Proxy集群间就会通过心跳的方式，把各自接收的客户端信息同步给彼此，同步客户端信息的时候，就会把客户端信息封装到RemoteChannel中
     * 然后把RemoteChannel的数据同步给其他Proxy节点
     */
    @Override
    public RemoteChannel toRemoteChannel() {
        return new RemoteChannel(
                //把proxy的网络地址设置进去
                ConfigurationManager.getProxyConfig().getLocalServeAddr(),
                //把当前proxy管理的channel对应的客户端网络地址设置进去
                this.getRemoteAddress(),
                //把本地地址设置进去
                this.getLocalAddress(),
                //设置channel协议类型
                ChannelProtocolType.GRPC_V2,
                //得到当前channel对应的客户端的设置信息
                this.getChannelExtendAttribute());
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：该类的对象表示GrpcChannel的Id，其实就是封装了channel对应的客户端的Id信息
     */
    protected static class GrpcChannelId implements ChannelId {

        private final String clientId;


        public GrpcChannelId(String clientId) {
            this.clientId = clientId;
        }


        @Override
        public String asShortText() {
            return this.clientId;
        }

        @Override
        public String asLongText() {
            return this.clientId;
        }


        @Override
        public int compareTo(ChannelId o) {
            if (this == o) {
                return 0;
            }
            if (o instanceof GrpcChannelId) {
                GrpcChannelId other = (GrpcChannelId) o;
                return ComparisonChain.start()
                        .compare(this.clientId, other.clientId)
                        .result();
            }
            return asLongText().compareTo(o.asLongText());
        }
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把可以和客户端通信的流式观察者对象设置到telemetryCommandRef成员变量中
     */
    public void setClientObserver(StreamObserver<TelemetryCommand> future) {
        this.telemetryCommandRef.set(future);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：清除流式观察者对象的方法
     */
    protected void clearClientObserver(StreamObserver<TelemetryCommand> future) {
        this.telemetryCommandRef.compareAndSet(future, null);
    }



    //以下几个方法都是netty中的方法，我就不添加注释了，大家肯定都很熟悉了
    @Override
    public boolean isOpen() {
        return this.telemetryCommandRef.get() != null;
    }


    @Override
    public boolean isActive() {
        return this.telemetryCommandRef.get() != null;
    }


    @Override
    public boolean isWritable() {
        return this.telemetryCommandRef.get() != null;
    }

    /* @方法描述：处理其他消息的方法，从名字上看不出这个方法的具体作用，现在我可以先跟大家解释一句，假如是当前GrpcClientChannel主动向客户端发送消息
     * 那么最终就是调用这个方法把消息发送给客户端，至于为什么是这样，接着往下看，很快大家就清楚了
     */
    @Override
    protected CompletableFuture<Void> processOtherMessage(Object msg) {
        //判断消息是否为TelemetryCommand类型，如果是该类型，则意味着确实是grpc客户端主动向服务端发送消息
        if (msg instanceof TelemetryCommand) {
            TelemetryCommand response = (TelemetryCommand) msg;
            //把消息发送给客户端
            this.writeTelemetryCommand(response);
        }
        //返回一个设置了完成状态的future对象，现在大家肯定很疑惑，刚才在grpcchannel管理器中不是有一个responseFuture对象吗？、
        //不是说服务端主动向客户端发送请求会创建要给对应的responseFuture对象等待响应到来吗？为什么这里没创建呢？请大家看下面几个方法就清楚了
        return CompletableFuture.completedFuture(null);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理获取消费者客户端运行信息请求的方法,这个方法会和下面的processConsumeMessageDirectly()方法放在一起解释，下面那个方法我就不添加详细注释了
     * 现在我想请大家先想一想，在上一版本实现生产者客户端的时候，我们是不是在ClientSessionImpl的onNext()方法中看到了这样一段代码：
     *     public void onNext(TelemetryCommand command) {
     *         //得到客户端的唯一Id
     *         final String clientId = sessionHandler.clientId();
     *         try {
     *             //根据遥测请求的类型判断要执行的操作
     *             switch (command.getCommandCase()) {
     *                 case SETTINGS: {
     *                     //执行到这里意味着服务端发送过来的是设置命令，也就是说服务端向客户端发送了一个TelemetryCommand请求
     *                     //这个请求中包含的是服务端向客户端发送的客户端设置信息，当客户端接收到这些信息后，要把从这些设置信息同步到自己本地
     *                     final Settings settings = command.getSettings();
     *                     LOGGER.info("Receive settings from remote, endpoints={}, clientId={}", endpoints, clientId);
     *                     //将服务端发送过来的设置信息同步到自己本地
     *                     sessionHandler.onSettingsCommand(endpoints, settings);
     *                     break;
     *                 }
     *                 //以下遥测请求类型在当前版本代码中用不上，所以我就先不添加注释了，大家感兴趣的话可以自己看看，内容都很简单
     *                 case RECOVER_ORPHANED_TRANSACTION_COMMAND: {
     *                     final RecoverOrphanedTransactionCommand recoverOrphanedTransactionCommand =
     *                             command.getRecoverOrphanedTransactionCommand();
     *                     LOGGER.info("Receive orphaned transaction recovery command from remote, endpoints={}, "
     *                             + "clientId={}", endpoints, clientId);
     *                     sessionHandler.onRecoverOrphanedTransactionCommand(endpoints, recoverOrphanedTransactionCommand);
     *                     break;
     *                 }
     *                 case VERIFY_MESSAGE_COMMAND: {
     *                     final VerifyMessageCommand verifyMessageCommand = command.getVerifyMessageCommand();
     *                     LOGGER.info("Receive message verification command from remote, endpoints={}, clientId={}",
     *                             endpoints, clientId);
     *                     sessionHandler.onVerifyMessageCommand(endpoints, verifyMessageCommand);
     *                     break;
     *                 }
     *                 case PRINT_THREAD_STACK_TRACE_COMMAND: {
     *                     final PrintThreadStackTraceCommand printThreadStackTraceCommand =
     *                             command.getPrintThreadStackTraceCommand();
     *                     LOGGER.info("Receive thread stack print command from remote, endpoints={}, clientId={}",
     *                             endpoints, clientId);
     *                     sessionHandler.onPrintThreadStackTraceCommand(endpoints, printThreadStackTraceCommand);
     *                     break;
     *                 }
     *                 default:
     *                     //这里就是未知的遥测请求类型，记录日志即可
     *                     LOGGER.warn("Receive unrecognized command from remote, endpoints={}, command={}, clientId={}",
     *                             endpoints, command, clientId);
     *             }
     *         } catch (Throwable t) {
     *             LOGGER.error("[Bug] unexpected exception raised while receiving command from remote, command={}, "
     *                     + "clientId={}", command, clientId, t);
     *         }
     *     }
     *
     * 在这段代码中case VERIFY_MESSAGE_COMMAND和 case PRINT_THREAD_STACK_TRACE_COMMAND分支我都没有给大家添加注释，跟大家现在根本用不到这些方法，这个大家应该还有印象吧？我们当时只知道客户端会从服务端接收TelemetryCommand
     * 请求，然后根据请求中具体的命令执行不同操作，上一章我们只实现了TelemetryCommand请求中设置Settings命令的功能，剩下的都没实现，那现在到了的第十版本代码，看到这里大家应该清楚遥测请求的PRINT_THREAD_STACK_TRACE_COMMAND命令
     * 和VERIFY_MESSAGE_COMMAND命令是在哪里设置的了吧？就是下面的两个方法。在下面两个方法中会构建PRINT_THREAD_STACK_TRACE_COMMAND命令和VERIFY_MESSAGE_COMMAND命令的TelemetryCommand遥测请求，然后把请求发送给客户端
     * 客户端接收到请求之后，就可以回复服务端响应了。比如说接收到PrintThreadStackTraceCommand命令就会把消费者客户端的运行信息收集起来响应给服务端，这个就是大概的流程，大家还不必先了解的那么细
     * 接下来我想跟大家说的是，虽然proxy服务端会向客户端发送各种命令的遥测请求，但是下面两个方法，实际上还有本类的processCheckTransaction方法，这三个方法中的请求都不是proxy服务端主动发送给客户端的
     * 实际上是这样的：是Broker想收集消费者客户端的运行信息，但是Broker又不和客户端直接联系，当然，如果使用的是旧版本的客户端，那么为了兼容旧版本客户端，Broker节点还是可以直接和客户端联系的。但现在我们只关注新版本的mq代码
     * 当broker节点不和客户端直接联系了，Broker又想知道消费者客户端的运行信息，那么Broker节点就会先向proxy发送一个GetConsumerRunningInfo请求，然后proxy节点会向客户端发送一个TelemetryCommand请求，这个请求中
     * 设置的就是PRINT_THREAD_STACK_TRACE_COMMAND命令，客户端接收到TelemetryCommand请求后就会把运行信息收集起来返回给proxy，然后Proxy就会把信息返回给Broker节点。在这个过程中，下面的这个方法之所以会被调用
     * 其实就是因为Broker节点想收集客户端的运行信息，Proxy为Broker节点发送给客户端的，现在大家应该清楚其中的流程了吧？当然，这只是一个大概流程，一会我们就可以从代码中一点点验证我所说的内容
     */
    @Override
    protected CompletableFuture<Void> processGetConsumerRunningInfo(RemotingCommand command,
                                                                    GetConsumerRunningInfoRequestHeader header,
                                                                    CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> responseFuture) {
        if (!header.isJstackEnable()) {
            return CompletableFuture.completedFuture(null);
        }//在发送消息给客户端之前要构建TelemetryCommand请求，在构建TelemetryCommand请求的过程中执行了
        //this.grpcChannelManager.addResponseFuture(responseFuture)这样一行代码，这行代码就把responseFuture对象添加到channel管理器的resultNonceFutureMap表中了
        //现在大家肯定很疑惑，这个responseFuture对象究竟是怎么出现的呢？在哪里创建的呢？这个问题的答案大家可以去当前类的父类ProxyChannel中寻找
        this.writeTelemetryCommand(TelemetryCommand.newBuilder()
                .setPrintThreadStackTraceCommand(PrintThreadStackTraceCommand.newBuilder()
                        //注意，在这里把与responseFuture响应对象对应的唯一标识设置到遥测请求中了，客户端在回复响应的时候，还会把这个唯一标识返回
                        .setNonce(this.grpcChannelManager.addResponseFuture(responseFuture))
                        .build())
                .build());
        return CompletableFuture.completedFuture(null);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理消费者客户端直接消费消息请求的方法
     */
    @Override
    protected CompletableFuture<Void> processConsumeMessageDirectly(RemotingCommand command,
                                                                    ConsumeMessageDirectlyResultRequestHeader header,
                                                                    MessageExt messageExt, CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> responseFuture) {
        this.writeTelemetryCommand(TelemetryCommand.newBuilder()
                .setVerifyMessageCommand(VerifyMessageCommand.newBuilder()
                        .setNonce(this.grpcChannelManager.addResponseFuture(responseFuture))
                        .setMessage(GrpcConverter.getInstance().buildMessage(messageExt))
                        .build())
                .build());
        return CompletableFuture.completedFuture(null);
    }

    public String getClientId() {
        return clientId;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把消息发送给客户端的方法
     */
    public void writeTelemetryCommand(TelemetryCommand command) {
        //得到流式观察者
        StreamObserver<TelemetryCommand> observer = this.telemetryCommandRef.get();
        if (observer == null) {
            log.warn("telemetry command observer is null when try to write data. command:{}, channel:{}", TextFormat.shortDebugString(command), this);
            return;
        }//加锁
        synchronized (this.telemetryWriteLock) {
            observer = this.telemetryCommandRef.get();
            if (observer == null) {
                log.warn("telemetry command observer is null when try to write data. command:{}, channel:{}", TextFormat.shortDebugString(command), this);
                return;
            }
            try {//在这里把请求发送给客户端
                observer.onNext(command);
            } catch (StatusRuntimeException | IllegalStateException exception) {
                log.warn("write telemetry failed. command:{}", command, exception);
                this.clearClientObserver(observer);
            }
        }
    }


    //和事务相关的方法，暂时不添加注释，等后面真的引入事务功能了再添加，我之所以这么早把这些不相关的方法引入进来，真的是因为怕以后忘了
    //请大家理解一下，现在直接忽略这些和当前版本代码内容无关的方法即可
    @Override
    protected CompletableFuture<Void> processCheckTransaction(CheckTransactionStateRequestHeader header,
                                                              MessageExt messageExt, TransactionData transactionData, CompletableFuture<ProxyRelayResult<Void>> responseFuture) {
        CompletableFuture<Void> writeFuture = new CompletableFuture<>();
        try {
            this.writeTelemetryCommand(TelemetryCommand.newBuilder()
                    .setRecoverOrphanedTransactionCommand(RecoverOrphanedTransactionCommand.newBuilder()
                            .setTransactionId(transactionData.getTransactionId())
                            .setMessage(GrpcConverter.getInstance().buildMessage(messageExt))
                            .build())
                    .build());
            responseFuture.complete(null);
            writeFuture.complete(null);
        } catch (Throwable t) {
            responseFuture.completeExceptionally(t);
            writeFuture.completeExceptionally(t);
        }
        return writeFuture;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("clientId", clientId)
                .add("remoteAddress", getRemoteAddress())
                .add("localAddress", getLocalAddress())
                .toString();
    }
}